package com.hps.shoping.good.component;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;

import java.nio.charset.Charset;
import java.util.concurrent.Executors;

/**
 * @author heps
 * @date 2019/9/1 10:00
 */
@Component
@Slf4j
public class MessageProducer {

    TransactionMQProducer producer;

    public MessageProducer() {
        producer = new TransactionMQProducer("stock_group");
        producer.setNamesrvAddr("192.168.117.139:9876");
        producer.setExecutorService(Executors.newFixedThreadPool(10));
        producer.setTransactionListener(new TransactionListenerImpl());

        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    private static class TransactionListenerImpl implements TransactionListener {

        @Override
        public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
            log.info("do update stock transaction {}", msg);
            return LocalTransactionState.UNKNOW;
        }

        @Override
        public LocalTransactionState checkLocalTransaction(MessageExt msg) {
            String body = new String(msg.getBody(), Charset.defaultCharset());
            log.info("check stock {}", body);
            return LocalTransactionState.COMMIT_MESSAGE;
        }
    }

    public void send(String message) throws Exception {
        Message msg = new Message("stockTopic", message.getBytes(Charset.defaultCharset()));
        SendResult sendResult = producer.sendMessageInTransaction(msg, null);
        log.info("update stock result {}", sendResult);
    }
}
